package com.opencv.thread;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.concurrent.*;

/**
 * @author qz.wu
 * @date 2022/1/20 9:19
 * @descriptions 池化思想：为了最大化收益并最小化风险，而将资源统一在一起管理的一种思想
 */
public class ThreadPool {

    // CPU密集型任务： N +1
    // I/O密集型任务： 2N

    private static final Logger LOGGER = LoggerFactory.getLogger(ThreadPool.class);

    /**
     * 线程池的核心线程数量
     */
    private final static int CORE_POOL_SIZE = Runtime.getRuntime().availableProcessors() * 2;

    /**
     * 线程池的最大线程数
     */
    private final static int MAXIMUM_POOL_SIZE = CORE_POOL_SIZE * 2;

    /**
     * 当线程数大于核心线程数时，多余的空闲线程存活的最长时间
     */
    private final static long KEEP_ALIVE_TIME = 15L;
    /**
     * 任务队列的总容量
     */
    private final static int QUEUE_CAPACITY = 100;

    /**
     * 异步任务线程池
     */
    private static final ThreadPoolExecutor ASYNC_TASK_POOL = new ThreadPoolExecutor(
            CORE_POOL_SIZE, MAXIMUM_POOL_SIZE, KEEP_ALIVE_TIME, TimeUnit.SECONDS,
            new ArrayBlockingQueue<>(QUEUE_CAPACITY),
            createThreadFactory("lite-task-async", false),
            // 拒绝策略：ignore And throw a RejectedExecutionException
            new ThreadPoolExecutor.AbortPolicy()
    );

    /**
     * 同步任务线程池
     */
    private static final ThreadPoolExecutor SYNC_TASK_POOL = new ThreadPoolExecutor(
            CORE_POOL_SIZE, MAXIMUM_POOL_SIZE, KEEP_ALIVE_TIME, TimeUnit.SECONDS,
            new ArrayBlockingQueue<>(QUEUE_CAPACITY),
            createThreadFactory("lite-task-sync", false),
            // 拒绝策略：ignore And throw a RejectedExecutionException
            new ThreadPoolExecutor.AbortPolicy()
    );

    /**
     * 提交同步任务到同步线程池
     *
     * @param callable
     * @param <T>
     * @return
     * @throws Exception
     */
    public static <T> T submitSyncTask(Callable<T> callable) throws Exception {
        try {
            Future<T> future = SYNC_TASK_POOL.submit(callable);
            return future.get();
        } catch (Exception ex) {
            throw ex;
        }
    }

    /**
     * 提交任务到异步线程池执行
     *
     * @param command
     */
    public static void executeAsyncTask(Runnable command) {
        ASYNC_TASK_POOL.execute(command);
    }

    /**
     * 提交任务到同步线程池执行
     *
     * @param common
     * @throws ExecutionException
     * @throws InterruptedException
     */
    public static void executeSyncTask(Runnable common) throws ExecutionException, InterruptedException {
        Future<?> future = SYNC_TASK_POOL.submit(common);
        future.get();
    }

    /**
     * 打印线程池的状态
     *
     * @param threadPool
     */
    public static void printThreadPoolStatus(ThreadPoolExecutor threadPool) {
        ScheduledExecutorService scheduledExecutorService = new ScheduledThreadPoolExecutor(1, createThreadFactory("print-thread-pool-status", false), new ThreadPoolExecutor.AbortPolicy());
        scheduledExecutorService.scheduleAtFixedRate(() -> {
            LOGGER.info("=========================");
            LOGGER.info("ThreadPool Size: [{}]", threadPool.getPoolSize());
            LOGGER.info("Active Threads: {}", threadPool.getActiveCount());
            LOGGER.info("Number of Tasks : {}", threadPool.getCompletedTaskCount());
            LOGGER.info("Number of Tasks in Queue: {}", threadPool.getQueue().size());
            LOGGER.info("=========================");
        }, 0, 1, TimeUnit.SECONDS);
    }

    /**
     * 创建线程池工厂
     *
     * @param threadName
     * @param daemon
     * @return
     */
    private static ThreadFactory createThreadFactory(String threadName, boolean daemon) {
        return threadFactory -> {
            Thread thread = new Thread(threadFactory, threadName);
            thread.setDaemon(daemon);
            return thread;
        };
    }


    public static ThreadPoolExecutor getAsyncTaskPool() {
        return ASYNC_TASK_POOL;
    }

    public static ThreadPoolExecutor getSyncTaskPool() {
        return SYNC_TASK_POOL;
    }
}
